package org.activequant.math.algorithms;

import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.PriorityBlockingQueue;

import org.activequant.core.util.IteratorHelper;

/**
 * Sorting utility: takes input as a set of Iterators (they are supposed to
 * output data in the ascending order, according to the comparator).
 * Merges these data streams together into one Iterator that also outputs the
 * data in the ascending order.<br/>
 * The MergeSortIterator&lt;T&gt; extends IteratorHelper&lt;T&gt;. Holds the following associated variables:
 * <ul>
 * <li>comparator(Comparator&lt;T&gt;)</li>
 * <li>checkOrder(boolean)</li>
 * <li>streams(Queue&lt;Stream&gt;)</li>
 * <li>sortBucket(PriorityQueue&lt;Stream&gt;)</li>
 * <li>serialNumber(long)</li>
 * </ul>
 * <b>History:</b><br>
 * - [09.11.2007] Created (Mike Kroutikov)<br>
 * - [23.05.2009] Delay fetching iterators till sort's hasNext() (Mike Kroutikov)<br>
 * 
 * @author Mike Kroutikov
 */
public class MergeSortIterator<T> extends IteratorHelper<T> {
	/**
	 * private final Comparator&lt;T&gt; comparator;
	 */
	private final Comparator<T> comparator;
	/**
	 * private boolean checkOrder = true;
	 */
	private boolean checkOrder = true;
	/**
	 * private final PriorityQueue&lt;Stream&gt; sortBucket = new PriorityQueue&lt;Stream&gt;();
	 */
	private final PriorityQueue<Stream> sortBucket = new PriorityQueue<Stream>();
	/**
	 * private final Queue&lt;Stream&gt; streams = new LinkedList&lt;Stream&gt;();
	 */
	//private final Queue<Stream> streams = new LinkedList<Stream>(); // for adding things
	private final Queue<Stream> streams = new PriorityBlockingQueue<Stream>(); 
	/**
	 * private long serialNumber = 0L;
	 */
	private long serialNumber = 0L;
	/**
	 * constructs a MergeSortIterator(MergeSortIterator&lt;T&gt; extends IteratorHelper&lt;T&gt;) using the given comparator(Comparator&lt;T&gt;) to set its
	 * associated comparator(Comparator&lt;T&gt;)
	 * @param comparator
	 */
	public MergeSortIterator(Comparator<T> comparator) {
		this.comparator = comparator;
	}
	/**
	 * sets the associated checkOrder(boolean) to false
	 */
	public void disableOrderCheck() {
		checkOrder = false;
	}
	/**
	 * 
	 * A private class, implements the Comparable&lt;Stream&gt; interface. Holds the following associated variables:
	 * <ul>
	 * <li>iterator(Iterator&lt;T&gt;)</li>
	 * <li>top(T)</li>
	 * <li>serialNo(long)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	private class Stream implements Comparable<Stream> {
		/**
		 * public final Iterator&lt;T&gt; iterator;
		 */
		public final Iterator<T> iterator; // where from take the events
		/**
		 * public T top = null;
		 */
		public T top = null; // top element
		/**
		 * public long serialNo = 0L;
		 */
		public long serialNo = 0L; // for sort stability
		/**
		 * constructs a Stream(implements Comparable&lt;Stream&gt;) using the given i(Iterator&lt;T&gt;) to set its associated iterator(Iterator&lt;T&gt;)
		 * @param i
		 */
		Stream(Iterator<T> i) {
			iterator = i;
		}
		/**
		 * compares this Stream object to the given other(Stream) by comparing their associated top(T) and serialNo(long) elements
		 */
		public int compareTo(Stream other) {
			int rc = comparator.compare(this.top, other.top);
			if(rc != 0) return rc;
			if(this.serialNo > other.serialNo) {
				return 1;
			} else if(this.serialNo < other.serialNo) {
				return -1;
			}
			return 0;
		}
	}
	/**
	 * <strong>1.</strong> constructs a Stream object using the given iterator(Iterator&lt;T&gt;) to set its associated iterator(Iterator&lt;T&gt;)<br/>
	 * <strong>2.</strong> adds the new Stream object to the associated streams(Queue&lt;Stream&gt;)
	 * @param iterator
	 */
	public void addIterator(Iterator<T> iterator) {
		streams.offer(new Stream(iterator));
	}
	/**
	 * this merge sorts and returns the correct next element from the iterators in the associated streams(Queue&lt;Stream&gt;)
	 */
	@Override
	protected T fetchNext() {

		Stream stream;
		while((stream = streams.poll()) != null) {
			if(stream.iterator.hasNext()) {
				T top = stream.iterator.next();
				if(stream.top != null && comparator.compare(stream.top, top) > 0 && checkOrder) {
					throw new IllegalStateException("data in source iterator is out of order: " + stream.top + " > " + top);
				}
				stream.top = top;
				stream.serialNo = serialNumber++;
				sortBucket.add(stream);
			}
		}
		
		stream = sortBucket.poll();
		if(stream == null) {
			return null; // no more
		}
		
		streams.offer(stream);
		
		return stream.top;
	}
	/**
	 * clears the associated streams(Queue&lt;Stream&gt;) and sortBucket(PriorityQueue&lt;Stream&gt;), and then sets the associated serialNumber(long) to 0
	 */
	public void clear() {
		streams.clear();
		sortBucket.clear();
		serialNumber = 0L;
	}
	/**
	 * returns the size of the associated streams(Queue&lt;Stream&gt;)
	 * @return
	 */
	public int getStreamsSize(){
		return streams.size();
	}
}